package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2Map {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("map")
      .setMaster("local")

    val sc = new SparkContext(conf)

    /**
      * map : 一行一行处理rdd中的数据
      */

    /**
      * 构建rdd的方法
      * 1、读取文件
      * 2、基于scala集合构建rdd
      *
      */

    //基于scala集合构建rdd
    val listRDD: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,8,9),4)

    println("分区数量为：" + listRDD.getNumPartitions)

    val mapRDD: RDD[Int] = listRDD.map(i => {
      i * 2
    })

    //mapRDD.foreach(println)

    /**
      * mapValues 处理KV格式中的values数据
      */

    //转转成KV格式
    val kvRDD: RDD[(Int, Int)] = mapRDD.map(i => (i,i))

    val mapValuesRDD: RDD[(Int, Int)] = kvRDD.mapValues(i => i+1)

    //mapValuesRDD.foreach(println)

    /**
      * mapPartitions: 一次处理一个分区的数据，返回值需要是一个迭代器
      * mapPartitionsWithIndex: 多了一个下标
      */

    val mapPartitionsRDD: RDD[Int] = listRDD.mapPartitions((inter: Iterator[Int]) =>{
      val list: List[Int] = inter.toList
      list.map(i => i*2+1).toIterator
    })

    //mapPartitionsRDD.foreach(println)

    val mapPartitionsWithIndexRDD: RDD[Int] = listRDD.mapPartitionsWithIndex{
      case (index: Int,iter: Iterator[Int]) =>
        println("分区编号：" + index)
        iter
  }

    mapPartitionsWithIndexRDD.foreach(println)

  }

}
